package com.me.study.javaCore.io;

import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.expression.StringValue;
import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.insert.Insert;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author ME
 * @date 2020/10/23
 */
public class BigFileReadTest {

    private static LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(3000);

    static {
        try {
            Class.forName("org.postgresql.Driver");
        } catch (ClassNotFoundException e) {
            System.err.println("Load PostgreSQL Driver failed");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        // VM options:
        // -Xms200M -Xmx200M -Xmn150M -XX:MaxMetaspaceSize=200m
        // Program argument:
        // "C:\Users\lenovo\Documents\WeChat Files\y616468159\FileStorage\File\2020-10\2020-10-22.log" "jdbc:postgresql://139.199.200.207:2323/eboos?currentSchema=crm_user" "enterprisedb" "edb@123456"

        System.out.println("补偿任务开始时间：" + LocalDateTime.now());
        ExecutorService service = Executors.newCachedThreadPool();
        service.submit(new Producer(args[0]));
        for (int i = 0; i < 8; i++) {
            service.submit(new Consumer(args));
        }
    }

    static class Producer implements Runnable {
        private String filePath;

        public Producer(String filePath) {
            this.filePath = filePath;
        }

        @Override
        public void run() {
            try {
                BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(filePath), StandardCharsets.UTF_8));

                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.contains("[DEBUG] INSERT INTO syn_room as t (room_guid")) {
                        queue.put(line.substring(29));
                    }
                }

                // 读取结束，加入终止标记
                queue.put("end");
                System.out.println("文件解析结束时间：" + LocalDateTime.now());
            } catch (InterruptedException | IOException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {
        private String[] args;

        public Consumer(String[] args) {
            this.args = args;
        }

        @Override
        public void run() {
            Connection conn = null;
            Statement stmt = null;
            PreparedStatement selectPstm = null;

            try {
                conn = DriverManager.getConnection(args[1], args[2], args[3]);
                String selectSql = "SELECT room_guid FROM syn_room  WHERE room_guid=?";
                selectPstm = conn.prepareStatement(selectSql);
                stmt = conn.createStatement();
                BufferedWriter insertWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream("insertErr.txt"), StandardCharsets.UTF_8));

                String value;
                String line;
                while (true) {
                    line = queue.take();

                    // 约定 `end` 为终止标记，检验读取到的值
                    if ("end".equals(line)) {
                        Thread.sleep(2000);
                        System.out.println("收到终止标记");
                        System.out.println("数据校验结束时间：" + LocalDateTime.now());
                        // 执行完主动退出进程
                        System.exit(0);
                    }

                    ResultSet rs;
                    try {
                        Insert parse = (Insert) CCJSqlParserUtil.parse(line);
                        value = ((StringValue) ((ExpressionList) parse.getItemsList()).getExpressions().get(0)).getValue();
                        selectPstm.setString(1, value);
                        rs = selectPstm.executeQuery();

                        if (!rs.next()) {// 存在
                            stmt.executeUpdate(line);
                        }
                    } catch (SQLException | JSQLParserException e) {
                        insertWriter.write(line);
                        insertWriter.newLine();
                    }
                }
            } catch (InterruptedException | SQLException | IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    stmt.close();
                    selectPstm.close();
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
